package com.ponhu.ea2.stockcenter.provider.support.service;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.google.common.collect.Maps;
import com.ponhu.ea2.stockcenter.properties.StockProperties;
import com.ponhu.ea2.stockcenter.provider.constants.CommonConstant;
import com.ponhu.ea2.stockcenter.provider.entity.StockChangeMqLogs;
import com.ponhu.ea2.stockcenter.provider.entity.StockInventoryDocOccupy;
import com.ponhu.ea2.stockcenter.provider.entity.StockInventoryReal;
import com.ponhu.ea2.stockcenter.provider.entity.StockInventoryUnreal;
import com.ponhu.ea2.stockcenter.provider.service.entity.StockChangeMqLogsService;
import com.ponhu.ea2.stockcenter.provider.service.entity.StockInventoryDocOccupyService;
import com.ponhu.ea2.stockcenter.provider.service.entity.StockInventoryRealService;
import com.ponhu.ea2.stockcenter.provider.service.entity.StockInventoryUnrealService;
import com.ponhu.ea2.stockcenter.provider.utils.CommonUtils;
import com.ponhu.ea2.stockcenter.provider.utils.DingTalkHandler;
import com.ponhu.ea2.stockcenter.provider.utils.ProBeanUtils;
import com.ponhu.ea2.stockcenter.provider.utils.RedisUtils;
import com.ponhu.ea2.stockcenter.vo.DingTalkContentVO;
import com.ponhu.ea2.stockcenter.vo.StockMqVO;
import com.rabbitmq.client.Channel;
import io.netty.util.concurrent.FastThreadLocal;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.BooleanUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.format.DateTimeFormatter;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;

@Slf4j
@Component
public class RabbitMQService {

    @Autowired
    private StockProperties stockProperties;
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private DingTalkHandler dingTalkHandler;
    @Autowired
    private FastThreadLocal<String> traceIdLocal;
    @Autowired
    private RedisSupportService redisSupportService;
    @Autowired
    protected StockInventoryRealService stockInventoryRealService;
    @Autowired
    private StockChangeMqLogsService stockChangeMqLogsService;
    @Autowired
    private StockInventoryUnrealService stockInventoryUnrealService;
    @Autowired
    private StockInventoryDocOccupyService stockInventoryDocOccupyService;

    public void saleStockChangeSendMsg(Collection<String> inventoryCodes) {
        int batchSize = 100;
        int unit = (inventoryCodes.size() + batchSize - 1) / batchSize; // 向上取整
        for (int i = 0; i < unit; i++) {
            List<String> batchList = inventoryCodes.stream().skip(i * batchSize).limit(batchSize).collect(Collectors.toList());
            saleStockChangeSendMsg(batchList, Maps.newHashMap());
        }
    }

    public void saleStockChangeSendMsg(Collection<String> inventoryCodes, Map<String, Integer> stockMap) {
        if (CollectionUtils.isEmpty(inventoryCodes)) return;

        // 校验销售库存变动是否发送MQ的配置开关
        if (!stockProperties.getEnablePushMqData()) return;

        // 先校验对应货品MQ缓存屏障是否存在
        if (stockProperties.getEnablePushMqBarrier()) {
            inventoryCodes = inventoryCodes.stream().filter(inventoryCode ->
                            !redisSupportService.bucketIsExists(RedisUtils.getRedisKey(CommonConstant.MQ_SALESTOCK_CODE_EXP, inventoryCode)))
                    .collect(Collectors.toList());
            if (CollectionUtils.isEmpty(inventoryCodes)) return;
        }

        List<StockInventoryReal> realStockList = stockInventoryRealService.getRealInventoryStockList(inventoryCodes);

        String exchange = CommonConstant.SALE_STOCK_TOPIC_EXCHANGE;
        String routingKey = CommonConstant.XCX_ROUTINGKEY;
        Map<String, StockChangeMqLogs> mqLogsMap = realStockList.stream().map(realStock ->
                        new StockChangeMqLogs().setInventoryCode(realStock.getInventoryCode())
                                .setStatus(CommonConstant.SEND_MQ_STATUS_0)
                                .setExchange(exchange).setRoutingKey(routingKey).setMessage("程序异常"))
                .collect(Collectors.toMap(StockChangeMqLogs::getInventoryCode, Function.identity()));
        try {
            Map<String, Integer> unrealMap = stockInventoryUnrealService.getUnRealInventoryStockList(inventoryCodes).stream()
                    .collect(Collectors.toMap(StockInventoryUnreal::getInventoryCode, StockInventoryUnreal::getSaleStockNum));

            DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
            List<StockMqVO> stockMqVOList = realStockList.stream()
                    .map(stock -> ProBeanUtils.copyProperties(stock, StockMqVO.class)
                            .setStock(stock.getMathSaleStockNum(unrealMap.get(stock.getInventoryCode())))
                            .setUpdatedTime(formatter.format(LocalDateTime.now())))
                    .filter(stock -> {
                        boolean hasChange = !stockMap.getOrDefault(stock.getInventoryCode(), Integer.MAX_VALUE).equals(stock.getStock());
                        if (BooleanUtils.isNotTrue(hasChange)) mqLogsMap.remove(stock.getInventoryCode());
                        return hasChange;
                    }).collect(Collectors.toList()); // filter：过滤掉未变化的库存信息

            if (CollectionUtils.isEmpty(stockMqVOList)) return;

            Message message = new Message(JSON.toJSONString(stockMqVOList).getBytes(), new MessageProperties());
            ReturnedMessage returnedMessage = new ReturnedMessage(message, 0, null, exchange, routingKey);
            CorrelationData correlationData = new CorrelationData();
            correlationData.setReturned(returnedMessage);
            rabbitTemplate.convertAndSend(exchange, routingKey, stockMqVOList, correlationData);
            log.info("货品库存销售库存变动发送消息程序成功 ===>> {} ===>> {} ===>> {}",
                    exchange, routingKey, JSON.toJSONString(stockMqVOList));

            String mapKey = RedisUtils.getRedisKey(CommonConstant.MQ_SALESTOCK_MAP, CommonUtils.getUUID());
            stockMqVOList.stream().forEach(mqVo -> {
                mqLogsMap.get(mqVo.getInventoryCode())
                        .setMessage(JSON.toJSONString(mqVo)).setStatus(CommonConstant.SEND_MQ_STATUS_1);

                if (stockProperties.getEnablePushMqBarrier() && CollectionUtils.isEmpty(stockMap)) {
                    redisSupportService.bucketSet(RedisUtils.getRedisKey(CommonConstant
                            .MQ_SALESTOCK_CODE_EXP, mqVo.getInventoryCode()), mapKey, stockProperties.getMqBarrierTime(), TimeUnit.SECONDS);
                    redisSupportService.bucketSet(RedisUtils.getRedisKey(CommonConstant
                            .MQ_SALESTOCK_CODE, mqVo.getInventoryCode()), mapKey);
                }
            });

            if (stockProperties.getEnablePushMqBarrier() && CollectionUtils.isEmpty(stockMap)) {
                Map<String, Integer> mqMap = stockMqVOList.stream()
                        .collect(Collectors.toMap(StockMqVO::getInventoryCode, StockMqVO::getStock));
                redisSupportService.rMapPutAll(mapKey, mqMap);
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("货品库存销售库存变动发送消息程序异常 ===>> {}", JSON.toJSONString(mqLogsMap.values()));
        } finally {
            List<StockChangeMqLogs> mqLogsList = mqLogsMap.values().stream()
                    .filter(logs -> logs.getStatus() == CommonConstant.SEND_MQ_STATUS_0).collect(Collectors.toList());
            if (CollectionUtils.isNotEmpty(mqLogsList)) {
                if (!stockChangeMqLogsService.saveBatch(mqLogsList))
                    log.error("货品库存销售库存变动发送消息保存日志异常 ===>> {}", JSON.toJSONString(mqLogsMap.values()));

                mqFailSendDingTalk(JSON.toJSONString(mqLogsList), "MQ发送程序异常");
            }
        }
    }

    public void saleStockChangeSendMsg(String orderNo) {
        LambdaQueryWrapper<StockInventoryDocOccupy> queryWrapper = Wrappers
                .lambdaQuery(StockInventoryDocOccupy.class)
                .eq(StockInventoryDocOccupy::getDocNo, orderNo);
        List<String> inventoryCodeList = stockInventoryDocOccupyService.list(queryWrapper)
                .stream().map(StockInventoryDocOccupy::getInventoryCode).collect(Collectors.toList());
        saleStockChangeSendMsg(inventoryCodeList);
    }

    public void mqFailSendDingTalk(String requestParams, String exceptionMsg) {
        String traceId = traceIdLocal.get();
        CompletableFuture.runAsync(() -> {
            DingTalkContentVO dingTalkContentVO = new DingTalkContentVO();
            dingTalkContentVO.setPath("RabbitMQService#saleStockChangeSendMsg");
            dingTalkContentVO.setTraceId(traceId);
            dingTalkContentVO.setRequestParams(requestParams);
            dingTalkContentVO.setExceptionMsg(exceptionMsg);
            dingTalkContentVO.setExceptionType("库存变动发送MQ消息失败");

            dingTalkHandler.sendMsg(dingTalkContentVO);
        });
    }

//    @RabbitListener(queues = {CommonConstant.TB_QUEUE})
    public void receive(Message message, Channel channel) throws Exception {
        try {
            long time = (long) (Math.random() * 10000);
            TimeUnit.MILLISECONDS.sleep(time);
            System.out.printf("\033[32m %s ===>> %s ===>> %s ===>> %s \033[0m\n",
                    Thread.currentThread().getName(), LocalTime.now(), time, new String(message.getBody()));
        } catch (Exception e) {
            e.getMessage();
        } finally {
            long deliverTag = message.getMessageProperties().getDeliveryTag();
            channel.basicAck(deliverTag, false);
        }
    }

}
